package com.ydj.service.impl;

import com.google.common.collect.Lists;
import com.ydj.service.MonitorService;
import org.apache.logging.slf4j.SLF4JLogger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Service
public class MonitorServiceImpl implements MonitorService {
    @Qualifier("monitorThreadPoolTaskExecutor")
    @Autowired
    private ThreadPoolTaskExecutor monitorThreadPoolExecutor;
    //创建日志对象
    private Logger logger = LoggerFactory.getLogger(MonitorServiceImpl.class);
    @Override
    public void monitor() {
        //第一布，查询异常数据,也就是需要监控的数据，数据量比较大
        List<String> dataList = initDataList();

        //第二部，将要查询出来的数据库进行分组
            //Lists:com.google.guava 每1000条数据进行分组
        List<List<String>> partitions =
                Lists.partition(dataList, 1000);
            //countDownLatch：基于计数器实现，等待子线程执行完成后，主线程才执行。构造参数为等待子线程的数量
        //计数器的值就是分组的数量，每个分组执行完后计数器的值减一，当计数器的值为0时，主线程才放行。
        CountDownLatch countDownLatch = new CountDownLatch(partitions.size());

        //统计异常数量
        AtomicInteger exceptionCount = new AtomicInteger();


        //第三布，循环前面分好的组，对每个子分组集合进行处理
        for (List<String> subList : partitions) {
            //第四布，针对每个子集合进行处理
            monitorThreadPoolExecutor.execute(()->{
                //第五步，统计异常数据
                try {
                    handle(exceptionCount,subList);
                }catch (Exception e){
                    logger.error("统计异常数据发生异常");
                }finally {
                    //计数器数量减一
                    countDownLatch.countDown();
                }
            });
        }
        //阻塞主线程
        try {
            //设置超时时间为5分钟,根据项目来定
            countDownLatch.await(5, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //拿到异常数据后进行其他操作，比如记录日志，保存到数据库，或者发送邮件警告
        System.out.println("异常数量:"+exceptionCount.get());
    }

    private void handle(AtomicInteger exceptionCount, List<String> subList) {
        for (String s : subList) {
            //需要根据业务的一些规则判断是否存在异常
            if (Integer.parseInt(s) >= 90000) {
                //模拟大于90000的数据是异常数据
                exceptionCount.incrementAndGet();
            }

        }
    }

    //初始化异常数据，真实场景查询数据库
    private List<String> initDataList() {
        List<String> list = new ArrayList<>();
        for (int i = 0; i < 100000; i++) {
            list.add(String.valueOf(i));
        }
        return list;
    }
}
